package cn.z2huo.demo.kafka.spring.consumer.consumer;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 *
 * @author z2huo
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class TestTopicKafkaConsumer {

    /*
    @Headers Map<String, String> headerMap
    使用 @Headers 接收到的 header map 示例如下
    {kafka_offset=6,
    kafka_consumer=org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer@1f0a3b84,

    {
        kafka_offset=6,
        kafka_consumer=org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer@1f0a3b84,
        kafka_timestampType=CREATE_TIME,
        kafka_receivedPartitionId=2,
        kafka_receivedTopic=test-topic-2,
        kafka_receivedTimestamp=1746012253659,
        kafka_acknowledgment=Acknowledgment for test-topic-2-2@6,
        kafka_groupId=testTopicKafkaListener
    }

     */

    @KafkaListener(id = "testTopicKafkaListener", containerFactory = "testTopicKafkaListenerContainerFactory",
            topics = "${spring.kafka.test-topic.topic-name}",
            autoStartup = "false",
            // batch 用于控制是否为 批处理 listener，批处理的话，方法参数中可以使用 list 来批量获取消息，但是批处理 listener 不支持 非阻塞重试
            batch = "false")
    public void listen(@Payload String message,
//                       @Headers Map<String, String> headerMap,
//                       @Header(KafkaHeaders.CONSUMER) KafkaConsumer consumer,
                       Acknowledgment acknowledgment) {

//        log.info("receive header is {}, receive message data is {}", headerMap, message);

        log.info("receive message data is {}", message);

        try {
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            acknowledgment.acknowledge();
        }
    }

    @KafkaListener(id = "testTopicBatchKafkaListener", containerFactory = "testTopicKafkaListenerContainerFactory",
            topics = "${spring.kafka.test-topic.topic-name}", clientIdPrefix = "batch-listener"
    )
    public void batchListen(@Payload List<String> messageList, Acknowledgment acknowledgment) {

//        log.info("receive header is {}, receive message data is {}", headerMap, message);

        System.out.println(Thread.currentThread().getName());

        for (int i = 0; i < messageList.size(); i++) {
            log.info("the NO.{} message in list, list size is {}, receive message data is {}",
                    i + 1, messageList.size(), messageList.get(i));
        }

        try {
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            acknowledgment.acknowledge();
        }
    }

}
